package com.huc.rabbit.mq.consumer;

import com.huc.rabbit.mq.config.ChannelConfig;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.GetResponse;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;

@Component
public class ChannelMessageConsumer {

    @Autowired
    private Channel channelConsumer;

    public void consumer() {
        new Thread(() -> {
            while (true) {
                try {
                    GetResponse response = channelConsumer.basicGet(ChannelConfig.QUEUE_NAME, false);
                    if (response == null) {
                        Thread.sleep(5000);
                        continue;
                    }
                    System.out.println("消费消息：" + new String(response.getBody()));
                    //手动回执确认消费成功并且让mq队列删除消息
                    channelConsumer.basicAck(response.getEnvelope().getDeliveryTag(), false);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();

    }

    /**
     * 批量消费
     **/
    public void batchConsumer() {
        System.out.println("批量消费开始.......");
        int batchSize = 5000;
        List<GetResponse> queues = new ArrayList<>();
        AtomicLong tag = new AtomicLong();
        new Thread(() -> {
            while (true) {
                try {
                    while (queues.size() < batchSize) {
                        GetResponse response = channelConsumer.basicGet(ChannelConfig.QUEUE_NAME, false);
                        if (response == null) {
                            break;
                        }
                        queues.add(response);
                        tag.set(response.getEnvelope().getDeliveryTag());
                    }

                    if(queues.isEmpty()) {
                        Thread.sleep(5000);
                    }else {
                        for (GetResponse response:queues) {
                            System.out.println("消费消息：" + new String(response.getBody()));
                        }
                        //手动回执确认消费成功并且让mq队列删除消息
                        channelConsumer.basicAck(tag.get(), false);
                        queues.clear();
                    }
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }).start();
    }

    public static void main(String[] args) throws IOException, TimeoutException {

        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setPort(5672);
//        connectionFactory.setVirtualHost("/");

        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();

        String exchangeName = "test_qos_exchange";
        String queueName = "test_qos_queue";
        String routingKey = "qos.#";

        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);

        //1 限流方式  第一件事就是 autoAck设置为 false

//        channel.basicQos(0,3,false);
//        String s = channel.basicConsume(queueName,false,new MyConsumer(channel));
//        System.out.println(s);

        while (true) {
            GetResponse response = channel.basicGet(queueName, false);
            if (response == null) {
                break;
            }
            System.out.println(new String(response.getBody()));
            channel.basicReject(response.getEnvelope().getDeliveryTag(), true);
//            channel.basicAck(response.getEnvelope().getDeliveryTag(),false);
        }
    }


//    private void consume(Channel channel) throws IOException, InterruptedException {
//        while (true) {
//            if (!isConditionSatisfied()) {
//                TimeUnit.MILLISECONDS.sleep(1);
//                continue;
//            }
//            GetResponse response = channel.basicGet(CAOSH_TEST_QUEUE, false);
//            if (response == null) {
//                TimeUnit.MILLISECONDS.sleep(1);
//                continue;
//            }
//            String data = new String(response.getBody());
//            logger.info("Get message <= {}", data);
//            channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
//        }
//    }
}
